package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

public class DimFuntion extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Map<String,TableProcess> startMap = new HashMap<>();

    public DimFuntion(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        String sql = "select * from edu_config.table_process_dim ";

        PreparedStatement preparedStatement = conn.prepareStatement(sql);

        ResultSet resultSet = preparedStatement.executeQuery();
        ResultSetMetaData metaData = preparedStatement.getMetaData();
        while (resultSet.next()){
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object value = resultSet.getObject(i);
                jsonObject.put(columnName,value);
            }
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            startMap.put(tableProcess.getSourceTable(),tableProcess);
        }
        resultSet.close();
        preparedStatement.close();
        conn.close();

    }

    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        String table = jsonObj.getString("table");
        TableProcess tableProcess = null;
        if ((tableProcess = broadcastState.get(table)) != null ||
                (tableProcess = startMap.get(table))!= null){

            JSONObject data = jsonObj.getJSONObject("data");

            String sinkColumns = tableProcess.getSinkColumns();

            filterColumn(data,sinkColumns);

            //加入其他字段
            data.put("TableProcess",tableProcess);
            data.put("type",jsonObj.getString("type"));
            out.collect(data);

        }


    }

    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] columns = sinkColumns.split(",");
        List<String> list = Arrays.asList(columns);
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(
                new Predicate<Map.Entry<String, Object>>() {
                    @Override
                    public boolean test(Map.Entry<String, Object> entry) {
                        return !list.contains(entry.getKey());
                    }
                }
        );

    }

    @Override
    public void processBroadcastElement(TableProcess orderBean, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        String op = orderBean.getOp();
        String sourceTable = orderBean.getSourceTable();
        if ("d".equals(op)){
            broadcastState.remove(sourceTable);
            startMap.remove(sourceTable);
        }else {
            broadcastState.put(sourceTable,orderBean);
            startMap.put(sourceTable,orderBean);
        }
    }
}
